package com.atguigu.gmall.app.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.util.MyConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;

import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * @author : ranzlupup
 * @since : 2023/6/1 19:50
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    /*Phoenix连接*/
    private Connection connection;
    private MapStateDescriptor<String, com.atguigu.gmall.bean.TableProcess> tableConfigDescriptor = null;

    public TableProcessFunction(MapStateDescriptor<String, com.atguigu.gmall.bean.TableProcess> tableConfigDescriptor) {
        this.tableConfigDescriptor = tableConfigDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 加载驱动
        Class.forName(MyConfig.PHOENIX_DRIVER);

        // 创建连接
        connection = DriverManager.getConnection(MyConfig.PHOENIX_SERVER);
    }

    /**
     * &#x5E7F;&#x64AD;&#x6D41;&#x6570;&#x636E;&#x5904;&#x7406;&#x65B9;&#x6CD5;
     *
     * @param jsonStr   广播流配置流中的每一条数据
     *                  {"before":null,
     *                  "after":{
     *                  "source_table":"financial_sku_cost",
     *                  "sink_table":"dim_financial_sku_cost",
     *                  "sink_columns":"id,sku_id,sku_name,busi_date,is_lastest,sku_cost,create_time",
     *                  "sink_pk":"id",
     *                  "sink_extend":null},
     *                  "source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1685621073781,"snapshot":"false","db":"gmall-config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1685621073781,"transaction":null}
     * @param context
     * @param collector
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(String jsonStr,
                                        BroadcastProcessFunction<JSONObject,
                                                String, JSONObject>.Context context,
                                        Collector<JSONObject> collector) throws Exception {
        //! 1.获取并解析数据，方便主流操作
        JSONObject jsonObject = JSONObject.parseObject(jsonStr);
        // String op = jsonObject.getString("op");
        // if ("d".equals(op)) {
        //     TableProcessFunction before = jsonObject.getObject("before", TableProcessFunction.class);
        //
        // }
        com.atguigu.gmall.bean.TableProcess tableProcess = jsonObject.getObject("after", com.atguigu.gmall.bean.TableProcess.class);
        String sinkTable = tableProcess.getSinkTable();
        String sinkColumns = tableProcess.getSinkColumns();
        String sinkPk = tableProcess.getSinkPk();
        String sinkExtend = tableProcess.getSinkExtend();

        //! 2.检验表是否存在，如果不存在则需要在Phoenix中建表 checkTable
        checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);

        //! 3.写入状态，广播出去
        BroadcastState<String, com.atguigu.gmall.bean.TableProcess> tableConfigState = context.getBroadcastState(tableConfigDescriptor);
        tableConfigState.put(tableProcess.getSourceTable(), tableProcess);
    }

    /**
     * 处理表数据，并建表
     *
     * @param sinkTable   Phoenix表名
     * @param sinkColumns Phoenix表字段
     * @param sinkPk      Phoenix表主键
     * @param sinkExtend  Phoenix表扩展字段
     */
    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        PreparedStatement preparedStatement = null;
        try {
            // 处理特殊字段
            // 主键字段
            if (sinkPk == null || "".equals(sinkPk)) {
                sinkPk = "id";
            }

            if (sinkExtend == null) {
                sinkExtend = "";
            }

            // 拼接SQL: create table if not exists db.table(id varchar primary key, col varchar, ...) xxx
            StringBuilder createTableSQL = new StringBuilder("create table if not exists ")
                    .append(MyConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");
            String[] columns = sinkColumns.split(",");
            for (int i = 0; i < columns.length; i++) {
                // 取出字段
                String column = columns[i];

                // 判断是否为主键
                if (sinkPk.equals(column)) {
                    createTableSQL.append(column).append(" varchar primary key");
                } else {
                    createTableSQL.append(column).append(" varchar");
                }
                // 判断是否为最后一个字段
                if (i < columns.length - 1) {
                    createTableSQL.append(",");
                }
            }
            createTableSQL.append(")").append(sinkExtend);

            // 编译SQL
            System.out.println("建表语句为: " + createTableSQL);

            preparedStatement = connection.prepareStatement(createTableSQL.toString());

            // 执行SQL，建表
            preparedStatement.execute();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("建表失败: " + sinkTable);
        } finally {
            // 释放资源
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 主流数据处理方法
     *
     * @param jsonObject      {"database":"gmall-211126-flink","table":"base_trademark","type":"update","ts":1652499176,"xid":188,"commit":true,"data":{"id":13,"tm_name":"atguigu","logo_url":"/bbb/bbb"},"old":{"logo_url":"/aaa/aaa"}}
     *                        {"database":"gmall-211126-flink","table":"order_info","type":"update","ts":1652499176,"xid":188,"commit":true,"data":{"id":13,...},"old":{"xxx":"/aaa/aaa"}}
     * @param readOnlyContext 只读
     * @param collector
     * @throws Exception
     */
    @Override
    public void processElement(JSONObject jsonObject,
                               BroadcastProcessFunction<JSONObject,
                                       String, JSONObject>.ReadOnlyContext readOnlyContext,
                               Collector<JSONObject> collector) throws Exception {
        //! 1.获取广播的配置数据
        ReadOnlyBroadcastState<String, com.atguigu.gmall.bean.TableProcess> broadcastState = readOnlyContext.getBroadcastState(tableConfigDescriptor);
        // 主流中获取到的表名
        String tableName = jsonObject.getString("table");

        // 通过主流中的表名 获取 广播配置流中的数据（可能为null，tableName为事实表，或者table_process配置维度表未配置）
        com.atguigu.gmall.bean.TableProcess tableProcess = broadcastState.get(tableName);
        if (tableProcess != null) {
            // 如果不为null才处理
            //! 2.过滤字段 filterColumn，从主流中过滤出配置流中的字段内容
            filterColumn(jsonObject.getJSONObject("data"), tableProcess.getSinkColumns());

            //! 3.补充SinkTable字段并写出到主流中
            //TODO 补充SinkTable字段并写出到主流中（数据格式）sinkTable ==> dim_ + source_table  || dim_base_trademark
            jsonObject.put("sinkTable", tableProcess.getSinkTable());
            collector.collect(jsonObject);
        } else {
            // 通过主流中的表名 获取 广播配置流中的数据
            // （可能为null，tableName为事实表，或者维度表未配置）
            System.out.println("找不到对应的Key(事实表或配置表未配置) -> " + tableName);
        }


    }

    /**
     * 过滤字段
     *
     * @param data        {"id":13,"tm_name":"atguigu","logo_url":"/bbb/bbb"}
     * @param sinkColumns "id,tm_name"
     */
    private void filterColumn(JSONObject data, String sinkColumns) {
        String[] columns = sinkColumns.split(",");
        List<String> columnsList = Arrays.asList(columns);

        Set<Map.Entry<String, Object>> entries = data.entrySet();

        // 遍历data中的所有字段，如果sinkColumns中没有的，就过滤掉（删除）
        // while (iterator.hasNext()) {
        //     Map.Entry<String, Object> next = iterator.next();
        //     if (!columnsList.contains(next.getKey())) {
        //         iterator.remove();
        //     }
        // }
        entries.removeIf(next -> !columnsList.contains(next.getKey()));
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }
}
